package com.camemax.utils;

import com.camemax.pojo.UniqueView;
import com.camemax.pojo.UserBehavior;
import javafx.beans.property.SimpleBooleanProperty;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;


public class UV_ProcessAllWindowFunction extends ProcessAllWindowFunction<UserBehavior, UniqueView, TimeWindow> {

    // Jedis客户端
    private Jedis jedis;

    // 自定义布隆过滤器
    private SimpleBloomFilter bloomFilter;

    @Override
    public void open(Configuration parameters) {
        jedis = new Jedis("172.28.40.190",16379);
        bloomFilter = new SimpleBloomFilter(1 << 29 ); // 处理一亿条数据 使用capacity=2^29的位图
    }

    @Override
    public void process(Context context, Iterable<UserBehavior> elements, Collector<UniqueView> out) throws Exception {
        // 时间窗口的TimeEnd为Key 作为count值计算的key
        long windowEnd = context.window().getEnd();
        String hashKey = String.valueOf(windowEnd);

        // 定义hash表"uv_count"， 用于redis中存放
        String redisHash = "uv_count";
        Long userId = elements.iterator().next().getUserId();
        Long offset = bloomFilter.hashCode(userId.toString(), 73);

        // 检索位图是否存在数据
        if (!jedis.getbit(hashKey, offset)) {
            jedis.setbit(hashKey, offset, true);

            String hashExists = jedis.hget(redisHash, hashKey);
            long count = 0L;
            if (StringUtils.isNotEmpty(hashExists) && !"".equals(hashExists)){
                // 获取当前hash表中offset的值，并加一后更新
                count = Long.parseLong(hashExists);
            }
            jedis.hset(redisHash,hashKey, String.valueOf(count+1));
            out.collect(new UniqueView("uv", windowEnd, (int)(count+1)));
        }
    }

    @Override
    public void close() throws Exception {
        jedis.close();
    }
}
